package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: kongcb
 * @Description: TODO
 * @Date: 2021/12/15 10:30
 */
public class TFinancing {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_financing").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_pevc_invevents");
        long zs_count = zs_data.count();
        if(zs_count>0){

            //企查查融资信息:t_financing获取统一社会信用代码
            spark.sql("select a.*,b.credit_code from a_dc_ep_dwi.t_financing a left join a_dc_ep_ods.t_eci_company b on a.company_id = b.company_id where length(b.credit_code)>5")
                    .createOrReplaceTempView("qcc_financing");
            //中数表取统一社会信用代码FINNAME_ENTID
            spark.sql("select a.* ,b.creditcode from a_dc_ep_ods.zs_t_pevc_invevents a left join a_dc_ep_ods.zs_t_pevc_invevents_del b on a.FINNAME_ENTID = b.FINNAME_ENTID where length(b.creditcode)>5")
                    .createOrReplaceTempView("zs_financing");


            //新增
            spark.sql("insert overwrite table a_dc_ep_incr.t_financing select " +
                    "zs.RECORD_ID," +
                    "qcc.key_no," +
                    "qcc.company_id," +
                    "qcc.company_name," +
                    "md5(zs.EVENTNAME)," +
                    "zs.INVDATE," +
                    "if(zs.CURNAME='其他',zs.INVAMT,if(zs.INVAMT!='',concat(zs.INVAMT,'万',zs.CURNAME),zs.INVAMT))," +
                    "zs.CUR," +
                    "zs.CURNAME," +
                    "zs.INVNATURE_NAME," +
                    "zs.INVNAME_ALL," +
                    "md5(zs.INVNAME)," +
                    "zs.INVNAME," +
                    "zs.EVENTNAME," +
                    "'' as news_url," +
                    "'' as valuation," +
                    "'' as ratio," +
                    "zs.jobid," +
                    "'1' as isadd " +
                    "from zs_financing zs " +
                    "left join a_dc_ep_ods.t_eci_company qcc on upper(trim(zs.creditcode)) = upper(trim(qcc.credit_code))");

            //and zs.DECLAREDATE = date_format(qcc1.date,'yyyyMMdd')
            //删除
            spark.sql("insert into table a_dc_ep_incr.t_financing select " +
                    "qcc.id,\n" +
                    "qcc.key_no,\n" +
                    "qcc.company_id,\n" +
                    "qcc.company_name,\n" +
                    "qcc.financing_id,\n" +
                    "qcc.date,\n" +
                    "qcc.amount,\n" +
                    "qcc.CUR," +
                    "qcc.CURNAME," +
                    "qcc.round,\n" +
                    "qcc.investment,\n" +
                    "qcc.product_id,\n" +
                    "qcc.product_name,\n" +
                    "qcc.news_title,\n" +
                    "qcc.news_url,\n" +
                    "qcc.valuation,\n" +
                    "qcc.ratio,\n" +
                    "qcc.dates," +
                    "'-1' as isadd " +
                    "from qcc_financing qcc where qcc.credit_code in (select del.creditcode from a_dc_ep_ods.zs_t_pevc_invevents_del del where length(del.creditcode)>5)");

        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_financing");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
       spark.stop();
    }
}
